#include "zmqClient.h"
#include "reaC++.h"
#include <QQueue>
#include <QJsonDocument>
#include <QNetworkProxy>

namespace rea {

zmqClient::zmqClient(const QJsonObject& aConfig) : QObject()
{
    connect(&ssdp_, SIGNAL(FoundServer(QString, QString, QString)), this, SLOT(ServerFound(QString, QString, QString)));
    connect(&search_timer_, &QTimer::timeout, [this](){
        ssdp_.StartDiscovery();
        rea::pipeline::run<QJsonObject>("clientBoardcast", rea::Json("value", "finding server..."));
    });

    auto send_server = [this](rea::stream<QJsonObject>* aInput){
        if (!m_valid)
            return;
        auto str = QString(QJsonDocument(aInput->data()).toJson());
        rea::pipeline::run("postZMQ", rea::Json("id", req + m_detail, "data", rea::JArray(str), "", false));
        aInput->out();
    };
    rea::pipeline::add<QJsonObject, rea::pipeDelegate>([this, send_server](rea::stream<QJsonObject>* aInput){
        send_server(aInput);
        tryFind(&m_calls, aInput->data().value("type").toString())->push_back(aInput->cache());
    }, rea::Json("name", "callServer", "delegate", "receiveFromServer"));
    rea::pipeline::add<QJsonObject, rea::pipePartial>(send_server, rea::Json("name", "postServer"));

    rea::pipeline::add<QJsonObject, rea::pipePartial>([](rea::stream<QJsonObject>* aInput){
        aInput->out();
    }, rea::Json("name", "receiveFromServer"));

    rea::pipeline::add<QJsonObject>([this](rea::stream<QJsonObject>* aInput){
        auto dt = aInput->data();
        if (dt.contains("ip") && dt.contains("port") && dt.contains("id"))
            ServerFound(dt.value("ip").toString(), dt.value("port").toString(), dt.value("id").toString());
        else
            tryConnectServer();
        aInput->out();
    }, rea::Json("name", "tryLinkServer"));

    rea::pipeline::add<QJsonObject>([](rea::stream<QJsonObject>* aInput){
        aInput->out();
    }, rea::Json("name", "clientBoardcast"));
}

zmqClient::~zmqClient(){
    search_timer_.stop();
}

void zmqClient::tryConnectServer()
{
    if (!search_timer_.isActive())
        search_timer_.start(1000);
}

void zmqClient::ServerFound(QString aIP, QString aPort, QString aID)
{
    m_detail = aIP + ":" + aPort + ":" + aID;
    rea::pipeline::run<QJsonObject>("clientBoardcast", rea::Json("value", "connect socket", "detail", m_detail));
    auto ip = "tcp://" + aIP + ":" + aPort,
            ip2 = "tcp://" + aIP + ":" + QString::number(aPort.toInt() - 1);
    auto soc = m_sockets.socket(req + m_detail);
    if (!soc){
        m_sockets.add(rea::Json(req + m_detail, rea::Json("type", "dealer", "ip", ip),
                                sub + m_detail, rea::Json("type", "sub", "ip", ip2, "filter", "dpst")));
        rea::pipeline::add<QJsonObject>([this](rea::stream<QJsonObject>* aInput){
            ReceiveMessage(aInput->data());
        }, rea::Json("name", "receiveZMQ_" + req + m_detail));

        rea::pipeline::add<QJsonObject>([this](rea::stream<QJsonObject>* aInput){
            ReceiveMessage(aInput->data());
        }, rea::Json("name", "receiveZMQ_" + sub + m_detail));

        rea::pipeline::add<bool>([this](rea::stream<bool>* aInput){
            ReceiveState(aInput->data() ? QAbstractSocket::SocketState::ConnectedState : QAbstractSocket::SocketState::UnconnectedState);
        }, rea::Json("name", "statusZMQ_" + req + m_detail));
    }
}

void zmqClient::ReceiveState(QAbstractSocket::SocketState aState){
    if (aState == QAbstractSocket::SocketState::ConnectedState){
        m_valid = true;
        auto dtl = m_sockets.socket(req + m_detail)->address;
        dtl.replace("tcp://", "");
        rea::pipeline::run<QJsonObject>("clientBoardcast", rea::Json("value", "socket is connected", "detail", dtl, "host", dtl));
        connected();
        search_timer_.stop();
    }else if (aState == QAbstractSocket::SocketState::UnconnectedState){
        m_valid = false;
        rea::pipeline::run<QJsonObject>("clientBoardcast", rea::Json("value", "socket is unconnected"));
        rea::pipeline::run<QJsonObject>("tryLinkServer", QJsonObject());
    }
}

void zmqClient::ReceiveMessage(const QJsonObject& aData)
{
    auto tp = aData.value("type").toString();
    rea::pipeline::run<QJsonObject>("clientBoardcast", rea::Json("value", "receive from server: " + tp));
    auto cls = tryFind(&m_calls, tp);
    if (cls->size() > 0){
        rea::pipeline::runC<QJsonObject>("receiveFromServer", aData, cls->front(), tp);
        cls->pop_back();
    }else
        rea::pipeline::run<QJsonObject>("receiveFromServer", aData, tp);
}

static rea::regPip<QJsonObject> init_server([](rea::stream<QJsonObject>* aInput) {
    static zmqClient clt;
}, rea::Json("name", "createZMQClient"));

}
